package com.bkhech.bigdata.tiktok_anchor.videoinfotop10;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;
import java.util.Objects;

/**
 * 数据指标统计作业
 * 需求：
 * 2：统计每天开播时长最长的前10名主播及对应的主播时长
 *
 * 分析：
 * 1: 为了统计每天开播最长的前10名主播信息，需要在map阶段获取数据中每个主播的 id 和直播时长
 * 2：所以 map 阶段的 <k2, v2> 为 <Text, LongWritable>
 * 3：在 reduce 端对相同主播的直播时长进行累加求和，把这些数据存储到一个临时的 map 容器集合中
 * 4：在 reduce 端的 cleanup 函数中对 map 容器集合中的数据根据直播时长进行排序
 * 5：最后在 cleanup 函数中把直播时长最长的前 10 名主播的信息写出到 hdfs 中
 *
 * @author guowm
 * @date 2022/7/6
 */

@Slf4j
public class VideoInfoJobTop10 {

    public static void main(String[] args) {
        Configuration conf = new Configuration();
        //指定我们使用的文件系统类型:
        conf.set("fs.defaultFS", "hdfs://hadoop01:9000");
        FileSystem fileSystem = null;
        try {
            fileSystem = FileSystem.get(conf);
            run(args, fileSystem);
            fileSystem.close();
        } catch (IOException e) {
            e.printStackTrace();
            if (Objects.nonNull(fileSystem)) {
                try {
                    fileSystem.close();
                } catch (IOException ex) {
                    throw new RuntimeException(ex);
                }
            }
        }
    }

    public static void run(String[] args, FileSystem fileSystem) {
        // 可以指定job需要配置的参数
        final Configuration conf = fileSystem.getConf();
        try {
            final String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
            if (otherArgs.length < 2) {
                System.err.println("Usage: VideoInfoJobTop10 <in> <out> <dt>");
                System.exit(2);
            }

            // 参数
            String inputDir = otherArgs[0];
            String outputDir = otherArgs[1];
            String dt = otherArgs[2];
            dt = DateUtils.transDataFormat(dt);
            conf.set(Cons.DT, dt);

            // 创建一个job
            final Job job = Job.getInstance(conf);
            // 注意：这一行必须设置，否则在集群中执行的是找不到 VideoInfoJobTop10 这个类
            job.setJarByClass(VideoInfoJobTop10.class);

            // 指定输入路径(可以是文件，也可以是目录)
            FileInputFormat.setInputPaths(job, new Path(inputDir));
            // 指定输出路径(只能指定一个不存在的目录)
            FileOutputFormat.setOutputPath(job, new Path(outputDir));

            // 指定map相关的代码
            job.setMapperClass(VideoInfoTop10Map.class);
            // 指定k2的类型
            job.setMapOutputKeyClass(Text.class);
            // 指定v2的类型
            job.setMapOutputValueClass(LongWritable.class);

            // 指定 reduce 相关代码
            job.setReducerClass(VideoInfoTop10Reduce.class);
            // 指定k3的类型
            job.setOutputKeyClass(Text.class);
            // 指定k3的类型
            job.setOutputValueClass(LongWritable.class);

            job.setNumReduceTasks(1);

            // 提交job
            final boolean waitForCompletion = job.waitForCompletion(true);
            log.info("---execute result: {}---", waitForCompletion);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
